package com.streamingkafka;

import com.util.JDBCUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.junit.Test;
import scala.Tuple2;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.*;

/**
 * @author Aunean
 * @date 2021/12/16
 */
public class StreamingKafka implements Serializable {

    @Test
    public void streaming_kafka() throws InterruptedException {
        SparkConf conf = new SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .setAppName("streaming_kafka").setMaster("local[2]");
        JavaStreamingContext context = new JavaStreamingContext(conf, Durations.seconds(1));

        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");
        kafkaParams.put("auto.offset.reset", "latest");
        kafkaParams.put("enable.auto.commit", false);

        Collection<String> topics = Arrays.asList("test1");
//        String[] topicArr = {"test1"};

        JavaInputDStream<ConsumerRecord<String, String>> dStream = KafkaUtils.createDirectStream(
                context,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams)
        );

//        dStream.print();
        JavaPairDStream<String, String> pairDStream = dStream.mapToPair(t -> {
            return new Tuple2<>(t.key(), t.value());
        });
        //pairDStream.print();//(1,click)

        JavaPairDStream<Tuple2<String, String>, Long> stream = pairDStream.countByValue();
        stream.print();
        stream.foreachRDD(new VoidFunction<JavaPairRDD<Tuple2<String, String>, Long>>() {
            @Override
            public void call(JavaPairRDD<Tuple2<String, String>, Long> tuple2LongJavaPairRDD) throws Exception {

                tuple2LongJavaPairRDD.foreach(new VoidFunction<Tuple2<Tuple2<String, String>, Long>>() {
                    @Override
                    public void call(Tuple2<Tuple2<String, String>, Long> t) throws Exception {
                        String sql = "update advert_click_count set count = count+" + t._2 + " where id = " + t._1._1;
                        JDBCUtils.executeUpdate(sql);
                    }
                });

            }
        });

        context.start();              // Start the computation
        context.awaitTermination();   // Wait for the computation to terminate
    }

    @Test
    public void producer() {

        //配置信息
        Properties pro = new Properties();
        //kafka服务器地址
        pro.put("bootstrap.servers", "master:9092");
        //设置数据key和value的序列化处理类
        pro.put("key.serializer", StringSerializer.class);
        pro.put("value.serializer", StringSerializer.class);

        //创建生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(pro);
        ProducerRecord<String, String> record = new ProducerRecord<>("test1", "广告1", "click");

        //发送记录
        producer.send(record);
        producer.close();
    }
}
